package spark.core.scala

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/2/3.
 */
object BroadcastWC {
  def main(args: Array[String]) {
    //1、构建spark上下文
    val conf = new SparkConf()
        .setAppName("BroadcastWC")
        .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    //2、读取数据生成RDD
    val path = "datas/halibote.txt"
    val rdd = sc.textFile(path)
    //构建了一个字典（添加我们想要去除的特殊字符）
    val list = List("��",",",".","'","*","?",":",";","-","”","“","’")
    //将list作为广播变量，广播出去！
    val broadcastList: Broadcast[List[String]] = sc.broadcast(list)
    //3、处理数据做wordcount
    //（原始）
  /*  val resultRdd =  rdd.flatMap(_.split(" "))
        .map(word => (word,1))
        .reduceByKey(_ + _)
*/
    /**
     * 因为单纯使用" "(空格)做切割，会出现很多乱七八糟的字符
     * 现在想要将这些字符过滤掉
     *
     * 方法一：当word包含特殊字符的时候，将这个word过滤掉
     */
   /* val resultRdd =  rdd.flatMap(_.split(" "))
      .filter(word => word.nonEmpty && !list.exists(ch => word.contains(ch)))
      .map(word => (word,1))
      .reduceByKey(_ + _)*/

    // 方法二：当word包含特殊字符的时候，将这个word的特殊字符过滤掉
    val resultRdd =  rdd.flatMap(_.split(" "))
          .filter(word => word.nonEmpty)
          .map(word =>{
      /**
       * list通过foldLeft做遍历，每一遍历都将word作为初始值传入
       * 将初始值做循环，判断初始值的每一个ch是否包含特殊字符，
       * 如果包含就将特殊字符替换成""(空字符串)
       * word是初始值
       * a是每一次遍历的word的其中一个字符串
       * b就是list中的特殊字符
       *
       */
      //调用广播变量，使用参数名称.value，直接调用广播变量
      //广播变量一旦广播出去，就不能修改他的值
      (broadcastList.value.foldLeft(word)((a,b) => a.replace(b,"")),1)
    })
    .reduceByKey(_ + _)

    //4、数据打印
    resultRdd.foreach(println)
  }
}
